package com.zzm.mqlisten;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.event.EventListener;
import org.springframework.stereotype.Component;

import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.exception.MQBrokerException;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.alibaba.rocketmq.remoting.exception.RemotingException;
import com.zzm.model.TUser;
import com.zzm.rocketmq.event.RocketMqEvent;
import com.zzm.service.TUserService;

/**
 * @Author:钟志苗
 * @Date: 8:35 2017/5/13
 * @Description:消费监听器
 */
@Component
public class ConsumerListen {
	private static Logger log = LoggerFactory.getLogger(ConsumerListen.class);

	@Autowired
	private TUserService tUserService;

	@EventListener(condition = "#event.topic=='test'")
	public void testListen(RocketMqEvent event) {
		DefaultMQPushConsumer consumer = event.getConsumer();
		try {
			String id = new String(event.getMessageExt().getBody(), "utf-8");
			Long sid = Long.valueOf(id);
			TUser tUser = new TUser();
			tUser.setId(sid);

//			boolean result = tUserService.saveTUser(tUser);
			boolean result = true;

			System.out.println("#event.topic=='test' -> " + result);
		} catch (Exception e) {
			//e.printStackTrace();
			System.out.println("消息处理失败m=" + event.getMessageExt());
			
			MessageExt msg = event.getMessageExt();
			
			if (msg.getReconsumeTimes() <= 1) {// 重复消费1次
				try {
					consumer.sendMessageBack(msg, 1);
					log.info("send expire msg back. topic={}, msgId={}, storeHost={}, queueId={}, queueOffset={}",
							msg.getTopic(), msg.getMsgId(), msg.getStoreHost(), msg.getQueueId(), msg.getQueueOffset());
				} catch (RemotingException | MQBrokerException | InterruptedException | MQClientException e1) {
					e1.printStackTrace();
					// 消息进行定时重试
				}
			} else {
				System.out.println("消息消费失败，定时重试");
			}
		}
	}

	/**
	 * 顺序消费
	 * 
	 * @param event
	 */
	@EventListener(condition = "#event.topic=='order'")
	public void normalListen(RocketMqEvent event) {
		try {
			System.out.println("顺序消息：" + new String(event.getMessageExt().getBody(), "utf-8"));
		} catch (Exception e) {
			e.printStackTrace();
		}

	}
}
